package com.infinitus.nginxInfo.service;



import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.infinitus.nginxInfo.config.KafkaConfig;
import com.infinitus.nginxInfo.service.util.SpringUtil;


public class KafkaSender {

	public static KafkaSender s_sender = null;

	private Producer<String, String> m_producer;

	private KafkaConfig m_configManager;
	

	public static KafkaSender getIntance() {
		synchronized (KafkaSender.class) {
			if (s_sender == null) {
				s_sender = new KafkaSender();
			}
		}
		return s_sender;
	}

	private KafkaSender() {
		try {
			m_configManager = SpringUtil.getBean("kafkaConfig",KafkaConfig.class);
			m_producer = new KafkaProducer<String, String>(m_configManager.getProperties(true, false));
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("error when init kakfa producer", e);
		}
	}

	public void send(String topic, String str) {
		m_producer.send(new ProducerRecord<String, String>(topic, str));
	}
	
}
